package com.vhall.component.service.room.event.handler.record;


import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.vhall.component.dao.room.RoomExtendsMapper;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.entity.room.entity.RoomExtendsEntity;
import com.vhall.component.plugin.common.annotation.RedisSpinLock;
import com.vhall.component.plugin.common.constant.RedisKey;
import com.vhall.component.plugin.core.service.PaasService;
import com.vhall.component.service.room.RoomCacheService;
import com.vhall.component.service.room.RoomInnerService;
import com.vhall.component.service.room.concant.RoomConstant;
import com.vhall.component.service.room.event.handler.AbstractEventHandler;
import com.vhall.component.service.room.event.handler.EventHandlerType;
import com.vhall.component.service.room.event.handler.RedirectUtil;
import com.vhall.component.service.room.manager.RoomMqManager;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import static com.vhall.component.service.room.event.EventConstant.VOD_TO_LIVE_END;

/**
 * @author jingwen.li
 * @date 2023/2/16
 */
@Slf4j
@Component
@EventHandlerType(VOD_TO_LIVE_END)
public class VODToLiveEndEventHandler extends AbstractEventHandler {
    @Autowired
    private RedirectUtil redirectUtil;
    @Value("${old.callback.record.url}")
    private String callbackUrl;
    @Autowired
    private RoomInnerService roomInnerService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RoomCacheService roomCacheService;

    @Autowired
    private RoomMqManager roomMqManager;

    @Autowired
    private RoomExtendsMapper roomExtendsMapper;

    @Resource
    private PaasService paasService;

    @Override
    public void handle(Map<String, Object> callBack) {
//        redirectUtil.redirect(callbackUrl, callBack);
        log.info("收到房间[{}]点播转直播播放结束事件：{}, 事件详情：{}", callBack.get("room_id").toString(), VOD_TO_LIVE_END, callBack);

        // {room_id=lss_e50f60ef, errno=200, datetime=2023-06-13 14:36:17, refer=vhall, signature=a8ffc1c8c019d069baa55dc2fef03888,
        // vod_id=7a853d65, task_id=5f734c37-6160-4d65-a0eb-bd3f73d79af0, time=1686638177, event=VODToLiveEnd, app_id=TVB9THkQ, repeat_times=3, status=1}


        String roomId = callBack.get("room_id").toString();

        RoomsEntityDTO roomsEntityDTO = roomInnerService.getRoomInfo(roomId);

        roomInnerService.updateByIl(RoomsEntityDTO.builder()
                .status(com.vhall.component.plugin.common.constant.RoomConstant.STATUS_STOP)
                .roomId(roomId).build());


        //直播结束之后 马上 将当前 直播间的连接数清理为0
        redisTemplate.opsForValue().set(RedisKey.CONNECT_COUNT_BY_ROOM + roomsEntityDTO.getIlId(), 0);

        //直播结束之后  将账号下面对应的 房间id  清除 set
        redisTemplate.opsForSet().remove(RedisKey.LIVING_ROOMS_OF_ACCOUNT + roomsEntityDTO.getAccountId(), roomsEntityDTO.getIlId());

        //将房间内的文档进行关闭
        redisTemplate.opsForZSet().removeRangeByScore(RedisKey.ROOM_DOC, roomsEntityDTO.getIlId(), roomsEntityDTO.getIlId());


        RoomExtendsEntity roomExtends = getRoomExtends(roomId);
        if (roomExtends != null) {
            roomExtends.setEndTime(LocalDateTime.now());
            roomExtends.setEndType(Integer.parseInt(com.vhall.component.plugin.common.constant.RoomConstant.HOST));
            roomExtendsMapper.updateById(roomExtends);
            //放入缓存
            roomCacheService.cacheRoomExtend(roomExtends);
            // 发送异步消息队列，处理上报数据
            roomMqManager.sendBigDataBroadcastRecordReportMessage(roomExtends.getId());
        }

        // 发送结束消息，延迟15秒
        stopLive(roomsEntityDTO);

    }

    private void stopLive(RoomsEntityDTO roomsEntityDTO) {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("send-message-live-over-%d").build();

        ScheduledThreadPoolExecutor executor1 = new ScheduledThreadPoolExecutor(1, namedThreadFactory);
        executor1.schedule(() -> {
            //====删除rtmp========
            String body = createEndLiveBody(roomsEntityDTO.getAccountId());
            //房间消息
            paasService.sendMessage(roomsEntityDTO.getChannelId(), body);
            //通知消息
            paasService.sendMessage(roomsEntityDTO.getNifyChannel(), body);

        }, 15, TimeUnit.SECONDS);
        executor1.shutdown();
    }

    private String createEndLiveBody(Integer accountId) {
        JSONObject json = new JSONObject();
        json.put("type", com.vhall.component.plugin.common.constant.RoomConstant.LIVE_OVER);
        json.put(RoomConstant.ROOM_JOIN_ID, accountId);
        return json.toJSONString();
    }

    /**
     * 查询房间额外信息
     *
     * @param roomId 房间id
     * @return 返回值
     */
    public RoomExtendsEntity getRoomExtends(String roomId) {
        String cacheKey = RedisKey.ROOM_EXTEND_CACHE + roomId;
        //从缓存获取 房间扩展信息
        RoomExtendsEntity roomExtendsEntity = roomCacheService.getRoomExtendCache(roomId);
        if (Objects.nonNull(roomExtendsEntity)) {
            return roomExtendsEntity;
        }
        return cacheAndHandleGetRoomExtend(roomId, cacheKey);
    }


    @RedisSpinLock(redisKey = "#cacheKey", expireTime = 10 * 1000)
    public RoomExtendsEntity cacheAndHandleGetRoomExtend(String roomId, String cacheKey) {
        //从缓存获取 房间扩展信息
        RoomExtendsEntity roomExtendsEntity = roomCacheService.getRoomExtendCache(roomId);
        if (Objects.isNull(roomExtendsEntity)) {
            LambdaQueryWrapper<RoomExtendsEntity> wrapper = Wrappers.lambdaQuery();
            wrapper.eq(RoomExtendsEntity::getRoomId, roomId);
            wrapper.eq(RoomExtendsEntity::getEndType, 0);
            wrapper.orderByDesc(RoomExtendsEntity::getCreatedAt);
            wrapper.last(" limit 1 ");
            roomExtendsEntity = roomExtendsMapper.selectOne(wrapper);
            if (Objects.isNull(roomExtendsEntity)) {
                return null;
            }
            //放入缓存
            roomCacheService.cacheRoomExtend(roomExtendsEntity);
        }
        return roomExtendsEntity;
    }
}
